package com.company.cloud.common.mq.redis.client.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import com.company.cloud.common.mq.redis.client.message.RedisMQMessage;

public class RedisMQProducer {
	
	private final static Logger logger = LoggerFactory.getLogger(RedisMQProducer.class);

	private StreamOperations<String, String, String> streamOperations;

	private static RedisMQProducer _install = new RedisMQProducer();

	private RedisMQProducer() {}

	public static RedisMQProducer getInstall() {
		return _install;
	}

	public void setStreamOperations(StreamOperations<String, String, String> streamOperations) {
		this.streamOperations = streamOperations;
	}
	
	private void validMessage(RedisMQMessage message) {
		if(message==null) {
			throw new RuntimeException("+++++ Redis MQ,message entity is null");
		}
		if(StringUtils.isEmpty(message.getTopic())) {
			throw new RuntimeException("+++++ Redis MQ,message topic is null");
		}
		if(CollectionUtils.isEmpty(message.getData())) {
			throw new RuntimeException("+++++ Redis MQ,message content is null");
		}
		
	}

	public String produce(RedisMQMessage message) {
		if(streamOperations==null) {
			throw new RuntimeException("+++++ MQ-Redis, redis config is null.");
		}
		this.validMessage(message);
//		MapRecord<String, String, String> record=MapRecord.create(message.getTopic(), message.getData());
//		streamOperations.add(record);
		RecordId id= streamOperations.add(message.getTopic(), message.getData());
		logger.debug("+++++ Redis MQ,add ...");
		return id.getValue();
	}
	
	

}
